package edu.nepu.flink.api.aggregation;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Date 2024/2/28 20:57
 * @Created by chenshuaijun
 */
public class SimpleOperator {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<WaterSensor> elementSource = env.fromElements(
                new WaterSensor("s1", 1L, 7),
                new WaterSensor("s1", 3L, 9),
                new WaterSensor("s2", 2L, 8),
                new WaterSensor("s2", 4L, 10),
                new WaterSensor("s3", 5L, 11)
        );

        /**
         * 所有的简单的数据聚合操作都要在keyedStream之下进行，其中包括：
         * sum:求和（相同的key）
         * min:求最小值（相同的key）,对于非对比字段，不进行更新，只保留相同key的第一条数据的值
         * max:求最大值（相同的key）,对于非对比字段，不进行更新，只保留相同key的第一条数据的值
         * minBy:在求最小值的时候更新费对比字段
         * maxBy:在求最大值的时候更新非对比字段
         */
        KeyedStream<WaterSensor, String> keyedStream = elementSource.keyBy(WaterSensor::getId);

        // 1-> sum: 我们对相同id的vc进行累加
//        keyedStream.sum("vc").print();

        // 2-> min: 对相同的key，我们保留vc更小的数据
//        keyedStream.min("vc").print();

        // 3-> max: 对于相同的key，我们保留vc最大的数据
//        keyedStream.max("vc").print();

        // 4-> minBy: 对相同的key，我们保留vc更小的数据,ts字段会变成最小vc值对应的waterSensor的ts值
//        keyedStream.minBy("vc").print();

        // 5-> maxBy: 对相同的key，我们保留vc更大的数据,ts字段会变成最大vc值对应的waterSensor的ts值
        keyedStream.maxBy("vc").print();
        env.execute();

    }
}
